package com.tcoiss.datafactory.service.impl;

import com.tcoiss.common.core.constant.Constants;
import com.tcoiss.common.core.enums.ErroEnums;
import com.tcoiss.common.core.exception.CustomException;
import com.tcoiss.common.core.utils.DateUtils;
import com.tcoiss.common.core.utils.StringUtils;
import com.tcoiss.common.datasource.annotation.DB;
import com.tcoiss.common.redis.service.RedisService;
import com.tcoiss.datafactory.config.DynamicJdbcTemplate;
import com.tcoiss.datafactory.domain.BusTable;
import com.tcoiss.datafactory.domain.BusTableColumn;
import com.tcoiss.datafactory.domain.DataSource;
import com.tcoiss.datafactory.domain.Test;
import com.tcoiss.datafactory.service.IDataSourceService;
import com.tcoiss.datafactory.service.IDynamicSqlService;
import com.tcoiss.datafactory.util.InsertRunable;
import com.tcoiss.datafactory.util.SQLTableUtil;
import com.tcoiss.datafactory.util.ThreadUncaughtExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.CallableStatementCallback;
import org.springframework.jdbc.core.CallableStatementCreator;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.object.BatchSqlUpdate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.util.*;
import java.util.Date;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Service
public class DynamicSqlServiceImpl implements IDynamicSqlService {
    private static final Logger log = LoggerFactory.getLogger(DynamicSqlServiceImpl.class);

    @Autowired
    private DynamicJdbcTemplate dynamicJdbcTemplate;

    @Autowired
    private IDataSourceService iDataSourceService;

    @Autowired
    private RedisService redisService;



    /**
     * 创建表
     * @param tabName 表名称
     * @param columns 表字段
     */

    public boolean createTable(String tabName,String dbName, List<BusTableColumn> columns) {

        DataSource dataSource = iDataSourceService.getDataSourceByDbName(dbName);
        String sql ;
        if(dataSource.getDatabaseType().equalsIgnoreCase("sqlserver")){
            sql = SQLTableUtil.createMSSQL(tabName,columns);
        }else{
            sql = genMysql(tabName,columns);
        }
        try {

            log.info("建表语句是："+sql);
            dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(sql);
            return true;
        } catch (Exception e) {
            log.info("建表失败" + e.getMessage());
            return false;
        }


    }

    private String genMysql(String tabName,List<BusTableColumn> columns) {
        String sql = "";
        String id = columns.get(0).getColumnName();


            if(!id.equals("id")){
                sql = "create table "+tabName+"( id bigint primary key AUTO_INCREMENT not null";
                if(columns!=null&&columns.size()>0){
                    sql+=",";

                    for(int i=0;i<columns.size();i++){
                        BusTableColumn column = columns.get(i);
                        //添加字段
                        sql+=column.getColumnName().trim()+" "+column.getColumnType()+" comment '"+column.getColumnComment()+"'";
                        //防止最后一个,
                        if(i<columns.size()-1){
                            sql+=",";
                        }
                    }
                }
            }else{
                sql = "create table "+tabName+"("+id+" bigint primary key not null";
                if(columns!=null&&columns.size()>0){
                    sql+=",";

                    for(int i=1;i<columns.size();i++){
                        BusTableColumn column = columns.get(i);
                        //添加字段
                        sql+=column.getColumnName().trim()+" "+column.getColumnType()+" comment '"+column.getColumnComment()+"'";
                        //防止最后一个,
                        if(i<columns.size()-1){
                            sql+=",";
                        }
                    }
                }
            }

            //拼凑完 建表语句 设置默认字符集
            sql+=")DEFAULT CHARSET=utf8mb4;";
            return sql;

    }

    private String genSqlServer(String tabName,List<BusTableColumn> columns) {
        String sql = "";
        String id = columns.get(0).getColumnName();


        if(!id.equals("id")){
            sql = "create table "+tabName+"( id bigint primary key AUTO_INCREMENT not null";
            if(columns!=null&&columns.size()>0){
                sql+=",";

                for(int i=0;i<columns.size();i++){
                    BusTableColumn column = columns.get(i);
                    //添加字段
                    sql+=column.getColumnName().trim()+" "+column.getColumnType();
                    //防止最后一个,
                    if(i<columns.size()-1){
                        sql+=",";
                    }
                }
            }
        }else{
            sql = "create table "+tabName+"("+id+" bigint primary key not null";
            if(columns!=null&&columns.size()>0){
                sql+=",";

                for(int i=1;i<columns.size();i++){
                    BusTableColumn column = columns.get(i);
                    //添加字段
                    sql+=column.getColumnName().trim()+" "+column.getColumnType();
                    //防止最后一个,
                    if(i<columns.size()-1){
                        sql+=",";
                    }
                }
            }
        }
        return sql;

    }

    /**
     * 创建视图
     * @param tabName 表名称
     * @param tab_fields 表字段
     */
    public boolean createView(String tabName,String dbName, List<String> tab_fields,String viewsql) {
        try {
            String fields = String.join(",",tab_fields);
            String sql = "create view "+tabName+"("+fields+")"+ " AS " + viewsql;
            log.info("创建视图语句是："+sql);
            dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(sql);
            return true;
        } catch (Exception e) {
            log.info("建表失败" + e.getMessage());
            return false;
        }
    }


    /**
     * 批量更新数据
     * @param busTable 表名
     * @param list 参数字段数据
     */
    @Override
    public int batchUpdateOrInsert(BusTable busTable, String[] tab_fields, List<Object[]> list)  {
        //先删除待更新的记录
        if(this.getCount(busTable.getBusTableName(),busTable.getDbName())>0){
            int[] rowNum = batchDelete(busTable,list);
            log.info("成功删除"+rowNum[0]+"条数据");
        }
        return this.batchInsert(busTable,tab_fields,list);


    }
    public int batchInsert(BusTable busTable,String[] tab_fields,List<Object[]> list)  {
        String sql = "insert into "+busTable.getBusTableName()+"("+ String.join(",",tab_fields)+") values(";
        for(int i=0;i<tab_fields.length;i++){
            sql+="?";
            //防止最后一个,
            if(i<tab_fields.length-1){
                sql+=",";
            }
        }
        sql+=");";
        log.info("添加数据的sql:"+sql);
        //执行
        long start = System.currentTimeMillis();
        int[] rowNum = dynamicJdbcTemplate.getJdbcTemplate(busTable.getDbName()).batchUpdate(sql,list);
        long end = System.currentTimeMillis();
        log.info("成功添加数据"+rowNum.length +"Finished. Time taken : " + (end - start)/1000 + " s.");
        return rowNum.length;
    }
    private void batchUpdate(String tabName,String dbName,String[] tab_fields,List<Object[]> list)  {
        String sql = "update "+tabName+" set";
        for(int i=0;i<tab_fields.length;i++){
            sql+=tab_fields[i];
            //防止最后一个,
            if(i<tab_fields.length-1){
                sql+="=?, ";
            }
        }
        sql+=";";
        log.info("更新数据的sql:"+sql);
        //执行
        int[] rowNum = dynamicJdbcTemplate.getJdbcTemplate(dbName).batchUpdate(sql,list);
        log.info("成功修改数据"+rowNum.length);
    }




    /**
     * 查询表 【查询结果的顺序要和数据库字段的顺序一致】
     * @param tabName 表名
     * @param whereFields 参数字段
     * @param data 参数字段数据
     * @param tab_fields 数据库的字段
     */
    public List<Map<String,Object>> query(String tabName,String dbName,String[] whereFields,String[] data,String[] tab_fields){
        //String[] result = null;
        String fields = String.join(",",tab_fields);
        String sql = "select "+fields+" from "+tabName+" where ";
        //添加ID过滤

        int length = whereFields.length;
        for(int i=0;i<length;i++){
            sql+=whereFields[i]+" = ? ";
            //防止最后一个,
            if(i<length-1){
                sql+=" and ";
            }
        }
        sql+=";";
        log.info("查询sql:"+sql);
        //查询结果集
        List<Map<String,Object>> results = dynamicJdbcTemplate.getJdbcTemplate(dbName).queryForList(sql,data);
        //存放结果集
        return results;
    }

    /**
     * 查询表 【查询结果的顺序要和数据库字段的顺序一致】
     * @param tabName 表名
     * @param whereSql 参数字段
     * @param tab_fields 数据库的字段
     */
    public List<Map<String,Object>> query(String tabName,String dbName,String whereSql,String[] tab_fields){
        //String[] result = null;
        String fields = String.join(",",tab_fields);
        String sql = "select "+fields+" from "+tabName;
        if(StringUtils.isNotBlank(whereSql)){
            sql = sql+whereSql;
        }
        sql+=";";
        log.info("查询sql:"+sql);
        //查询结果集
        List<Map<String,Object>> results = dynamicJdbcTemplate.getJdbcTemplate(dbName).queryForList(sql);
        //存放结果集
        return results;
    }
    /**
     * 查询表 【查询结果的顺序要和数据库字段的顺序一致】
     * @param tabName 表名
     * @param id 参数字段数据
     * @param tab_fields 数据库的字段
     */
    public Object queryById(String tabName,String dbName,long id,String[] tab_fields){
            String fields = String.join(",",tab_fields);
            String sql = "select "+fields+" from "+tabName+" where id = ? ;";
            log.info("查询sql:"+sql);
            //查询结果集
            Object result = dynamicJdbcTemplate.getJdbcTemplate(dbName).queryForObject(sql,
                    new RowMapper(){
                public Object mapRow(ResultSet rs, int rowNumber) throws SQLException {
                    Object[] objects = new Object[tab_fields.length];
                    for(int j=0;j<tab_fields.length;j++){
                        objects[j] = rs.getString(j+1);
                    }
                    return objects;
                }
            },new Object[]{id});
            //存放结果集
            return result;
    }

    /**
     * 查询最新时间表 【查询结果的顺序要和数据库字段的顺序一致】
     * @param tabName 表名
     * @param time_field 参数字段数据
     */
    public Date queryNewestTime(String tabName,String dbName,String time_field){
        String sql = "select max("+time_field+") modify_time from "+tabName;
        log.info("查询sql:"+sql);
        //查询结果集
        Date result = dynamicJdbcTemplate.getJdbcTemplate(dbName).queryForObject(sql, Date.class);
        //存放结果集
        return result;
    }


    /**
     * 获取某张表总数
     * @param tabName
     * @return
     */
    public Integer getCount(String tabName,String dbName){
        String sql = "select count(*) from "+tabName+" ;";
        return dynamicJdbcTemplate.getJdbcTemplate(dbName).queryForObject(sql,Integer.class);
    }

    /**
     * 获取某张表某个时间段的总数
     * @param table
     * @return
     */
    @Override
    public Long getCountByDateTime(BusTable table, String dbName){
        String sql;
        if(table.getBusTableName().equals("kd_flow_history")){
            sql = "SELECT count(*) from "+table.getBusTableName()+" where jz_date BETWEEN '"+DateUtils.getBeginDayOfYesterday(DateUtils.YYYY_MM_DD)+"' AND '"+ DateUtils.dateTimeNow(DateUtils.YYYY_MM_DD)+"' ;";
            log.info("查询数据的sql:"+ sql);
            return dynamicJdbcTemplate.getJdbcTemplate(table.getDbName()).queryForObject(sql,Long.class);
        }else if(table.getTableType().equals("bill")||table.getTableType().equals("base")) {//单据表
            sql = "SELECT count(*) from "+table.getBusTableName()+" where modify_time BETWEEN '"+DateUtils.getBeginDayOfYesterday(DateUtils.YYYY_MM_DD)+"' AND '"+ DateUtils.dateTimeNow(DateUtils.YYYY_MM_DD)+"' ;";
            return dynamicJdbcTemplate.getJdbcTemplate(dbName).queryForObject(sql,Long.class);
        }
        return null;
    }

    /**
     * 后台分页显示
     * @param tabName
     * @param pageNo
     * @param pageSize
     * @param tab_fields
     * @return
     */
    /*public List<String[]> queryForPage(String tabName,int pageNo,int pageSize ,String[] tab_fields){
        List<String[]> list = new ArrayList<String[]>();
        try {
            String sql = "select * from "+tabName+" LIMIT ?,? ; ";
            System.out.println("查询sql:"+sql);
            //预处理SQL 防止注入
            ps = conn.prepareStatement(sql);
            //注入参数
            ps.setInt(1,pageNo);
            ps.setInt(2,pageSize);
            //查询结果集
            ResultSet rs = ps.executeQuery();
            //存放结果集
            while(rs.next()){
                String[] result = new String[tab_fields.length];
                for (int i = 0; i < tab_fields.length; i++) {
                    result[i] = rs.getString(tab_fields[i]);
                }
                list.add(result);
            }
            //关闭流
            rs.close();
            ps.close();
            conn.close();  //关闭数据库连接
        } catch (SQLException e) {
            System.out.println("查询失败" + e.getMessage());
        }
        return list;
    }*/
    /**
     * 删除表
     * @param tabName 表名称
     */
    public boolean delete(String tabName,String dbName){
        try {
            String sql = "delete from "+tabName+";";
            log.info("删除数据的sql:"+sql);
            dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(sql);
            return true;
        } catch (Exception e) {
            log.info("删除数据失败" + e.getMessage());
            return false;
        }
    }

    /**
     * 按条件删除对应数据
     * @param tabName 表名
     * @param field 查询字段
     * @param fieldValue  查询字段条件
     * @return
     */
    @Override
    public boolean deleteByField(String tabName,String dbName,String field ,String fieldValue){
        try {
            String sql = "delete from "+ tabName + " where "+ field +" = '" + fieldValue + "';";
            System.out.println("删除数据的sql:"+sql);
            dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(sql);
            return true;
        } catch (Exception e) {
            System.out.println("删除数据失败" + e.getMessage());
            return false;
        }
    }

    /**
     * 清空表数据 假如集成表为基础资料则清空做全量同步 为业务单据时则根据kd_id删除数据，业务报表是则按日期清空数据
     * @param table 表名称
     */
    public int[] batchDelete(BusTable table,List<Object []> list){
        String sql = "";
        if(table.getBusTableName().equals("kd_flow_history")){
            sql = "delete from "+table.getBusTableName()+" where jz_date BETWEEN '"+DateUtils.getBeginDayOfYesterday(DateUtils.YYYY_MM_DD)+"' AND '"+ DateUtils.dateTimeNow(DateUtils.YYYY_MM_DD)+"' ;";
            log.info("删除数据的sql:"+sql);
            return dynamicJdbcTemplate.getJdbcTemplate(table.getDbName()).batchUpdate(sql);
        }else if(table.getTableType().equals("bill")||table.getTableType().equals("base")){//单据表
            List<Object[]> ids = new ArrayList<>();
            final long[] ids2 = new long[list.size()];
            for(int i=0;i<list.size();i++){
                long id = Long.parseLong((list.get(i))[0].toString());
                Object[] objectId = new Object[1];
                objectId[0] = id;
                ids.add(objectId);
                ids2[i] = id;
            }

            sql = "delete from "+table.getBusTableName()+" where kd_id= ? ;";
            log.info("删除业务单据数据的sql:"+sql);
//            BatchSqlUpdate bsu = new BatchSqlUpdate(dynamicJdbcTemplate.getJdbcTemplate(table.getDbName()).getDataSource(), sql);
//            bsu.setBatchSize(1000);
//            bsu.setTypes(new int[]{Types.BIGINT});
//            int[] num = new int[ids2.length];
//            for(int i = 0; i < ids2.length; ++i){
//                num[i] = bsu.update(new Object[]{ids2[i]});
//            }
//            bsu.flush();
//            return num;

            return dynamicJdbcTemplate.getJdbcTemplate(table.getDbName()).batchUpdate(sql,
            new BatchPreparedStatementSetter(){

                public void setValues(PreparedStatement ps, int i)

                        throws SQLException {

                    ps.setLong(1, ids2[i]);
                }

                public int getBatchSize() {

                    return ids2.length;

                }

            });
        }
        return new int[1];

    }


    /**
     * 判断表是否存在
     * @param tabName
     * @return
     */
    public boolean exitTable(String tabName,String dbName){
        try {
            String sql = "select count(*) from "+tabName+";";
            dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(sql);
            return true;
        } catch (Exception e) {
            log.info("表"+tabName+"不存在");
            return false;
        }

    }
    /**
     * 删除数据表
     * 如果执行成功则返回false
     * @param tabName
     * @return
     */
    public boolean dropTable(String tabName,String dbName,String tableType){
        try {
            String sql = "drop ";
            if("view".equals(tableType)){
                sql = sql+ " view "+tabName+";";
            }else{
                sql = sql+ " table "+ tabName+";";
            }
            dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(sql);
            return true;
        } catch (Exception e) {
            log.info("删除数据失败" + e.getMessage());
            return false;
        }

    }

    /**
     * 执行存储过程 有返回值的存储过程（非结果集）
     * 如果执行成功则返回成功条数
     * @param sql
     * @return
     */
    @Override
    public int callProcedureByParam(String sql,String dbName,Object[] objects) {

            return (int) dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(
                new CallableStatementCreator() {
                    public CallableStatement createCallableStatement(Connection con) throws SQLException {
                        CallableStatement cs = con.prepareCall(sql);
                        for (int i=0;i<objects.length;i++){
                            cs.setString(1+i, objects[i].toString());// 设置输入参数的值
                        }
                        cs.registerOutParameter(objects.length,Types.INTEGER);// 注册输出参数的类型
                        return cs;
                    }
                }, new CallableStatementCallback() {
                    public Object doInCallableStatement(CallableStatement cs) throws SQLException, DataAccessException {
                        cs.execute();
                        return cs.getInt(objects.length);// 获取输出参数的值*/
                    }
                });
    }

    /**
     * 执行存储过程 无返回值的存储过程（非结果集）
     * 如果执行成功则返回成功条数
     * @param sql
     * @return
     */
    @Override
    public int callProcedure(String sql,String dbName,Object[] objects) {
        for(Object object:objects){
            sql = sql.replace("?",object.toString());
        }
        dynamicJdbcTemplate.getJdbcTemplate(dbName).execute(sql);
        log.info("调用存储过程成功");
        return 1;
    }


//    public  void BulkToDB(DataTable dt,String dbName) throws SQLException {
////        SqlConnection sqlConn = new SqlConnection(
////                ConfigurationManager.ConnectionStrings["ConnStr"].ConnectionString);
////
////        SqlBulkCopy bulkCopy = new SqlBulkCopy(sqlConn);
////        bulkCopy.DestinationTableName = "BulkTestTable";
////        bulkCopy.BatchSize = dt.Rows.Count;
//        SqlConnection sqlConn = dynamicJdbcTemplate.getJdbcTemplate(dbName).getDataSource().getConnection();
//
//        try
//        {
//            sqlConn.Open();
//            if (dt != null && dt.Rows.Count != 0)
//                bulkCopy.WriteToServer(dt);
//        }
//        catch (Exception ex)
//        {
//            throw ex;
//        }
//        finally
//        {
//            sqlConn.Close();
//            if (bulkCopy != null)
//                bulkCopy.Close();
//        }
//    }

    public void insertThreadData(String tabName,String dbName, String[] tab_fields, List<Object[]> list) {
        int count = 40000;                   //一个线程处理300条数据
        int listSize = list.size();        //数据集合大小
        int runSize = (listSize/count)+1;  //开启的线程数
        List<Object[]> newlist = null;       //存放每个线程的执行数据
        ExecutorService executor = Executors.newFixedThreadPool(runSize);
        //创建两个个计数器
        CountDownLatch begin = new CountDownLatch(1);
        CountDownLatch end = new CountDownLatch(runSize);
        String sql = "insert into "+tabName+"("+ String.join(",",tab_fields)+") values(";
        for(int j=0;j<tab_fields.length;j++){
            sql+="?";
            //防止最后一个,
            if(j<tab_fields.length-1){
                sql+=",";
            }
        }
        sql+=");";
        log.info("添加数据的sql:"+sql);
        for (int i = 0; i < runSize ; i++) {
            //计算每个线程执行的数据
            if((i+1)==runSize){
                int startIndex = (i*count);
                int endIndex = list.size();
                newlist= list.subList(startIndex, endIndex);
                log.info("第"+i+"newlist:最后一个元素"+newlist.get(newlist.size()-1)[0].toString());
            }else{
                int startIndex = (i*count);
                int endIndex = (i+1)*count;
                newlist= list.subList(startIndex, endIndex);
                log.info("第"+i+"newlist:最后一个元素"+newlist.get(newlist.size()-1)[0].toString());
            }

            //线程类
            InsertRunable runable = new InsertRunable(dynamicJdbcTemplate.getJdbcTemplate(dbName),redisService,sql,newlist,begin,end,i);
            Thread.setDefaultUncaughtExceptionHandler(new ThreadUncaughtExceptionHandler());
            //这里执行线程的方式是调用线程池里的executor.execute(mythead)方法。
            executor.execute(runable);

        }
        //关闭线程池
        executor.shutdown();
        Long startTime = System.currentTimeMillis();
        try {
            // 请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行
            // 设置最长等待10秒
            executor.awaitTermination(30*runSize, TimeUnit.SECONDS);
            log.info("子线程执行结束");
            //break;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String ex = redisService.getCacheObject("ThreadException");
        if(StringUtils.isNotEmpty(ex)){
            throw new CustomException(ErroEnums.SYSERRO.getCode(),null,ex);
        }
//        while (Thread.activeCount()>1) {//等待所有任务都执行结束
//            Thread.yield();
//        }
        log.info("共耗时:"+(System.currentTimeMillis()-startTime)/1000.0+"s");
    }






}
